
                /*********************************************
                **                                          **
                **                                          **
                **                 ca.poll.C                **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **                12/31/97                  **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/

/*
Asynchronous Communication, direct polling.
Needs exactly three processes to execute.
*/


#include "arch.h"

//user program
Global_id gid1;
Coll_id cid1, cid2;

void init_gid(int, char **){
   gid1.set();
   cid1.set();
   cid2.set();
   fprintf(Sdf::trace, "node%d: gid1.tag=%d, cid1.tag=%d, cid2.tag=%d\n",
           Sdf::PROC_ID, gid1.get(), cid1.get(), cid2.get());
}

class SR_count{

   public:
      int cnt;

      SR_count(){
         cnt = 0;
      }
      void reset(){

         cnt = 0;
      }
      void more_SR(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();
#define not_complete  (c.cnt >0)

ChanA<int> *chan;

void upon_end_of_SR_r(ChanAs *chan){           
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_r, capa=%d, rcvd size=%d, Tsize=%d\n", 
           Sdf::PROC_ID, chan->get_chan_capacity(), 
           chan->get_msg_size(Sdf::RCV), chan->get_Tsize());
   fflush(Sdf::trace);
   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::RCV)))->cnt--;
}

void upon_end_of_SR_s(ChanAs *chan){           
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_s, capa=%d, sent size=%d, Tsize=%d\n", 
          Sdf::PROC_ID, chan->get_chan_capacity(), 
          chan->get_msg_size(Sdf::SND),  chan->get_Tsize());
   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::SND)))->cnt--;
   fflush(Sdf::trace);
   chan->get_Tsize();
}


class  Root_proc: public Thread{

   int recvd;

  public:
        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body){}

        void body(){

            fprintf(Sdf::trace, "node%d, in Root_proc::body1\n", 
                    Sdf::PROC_ID);
            fflush(Sdf::trace);
            chan = new ChanA<int>(gid1, (Sdf::PROC_ID+1)%Sdf::PROC_NB);

            fprintf(Sdf::trace, "node%d, in Root_proc::body2\n", 
                    Sdf::PROC_ID);
            fflush(Sdf::trace);

            Sdf::barrier(cid1);
                //synchronizes chan construction (no longer
                //necessary starting from version 5)

            SR_count c;

            chan->set(Sdf::SND, upon_end_of_SR_s, &c);
            chan->set(Sdf::RCV, upon_end_of_SR_r, &c);

            fprintf(Sdf::trace, "node%d, in Root_proc::body3\n", 
                    Sdf::PROC_ID);
            fflush(Sdf::trace);

            int sum_recvd = 0;
            for(int i=0; i<5; i++){
             int sent_data = Sdf::PROC_ID*10 + i;
             c.more_SR(2);
             chan->send(Sdf::STD, sent_data);
             chan->recv(recvd);
             wait_while(not_complete);
             Sdf::barrier(cid2);
               //global completion (actually not necessary since
               //since allreduce below cannot consume recvd before
               //it is locally available)

              //in single threaded codes MPI collective functions
              //can be used (as well as ARCH collective functions, 
              //see ca.cml, ca.poll).
             int reduce_out;
             MPI_Allreduce(&recvd, &reduce_out, 1, MPI_INT,
                           MPI_SUM, MPI_COMM_WORLD);
             sum_recvd += reduce_out;

              fprintf(Sdf::trace, "node%d, i = %d, recvd = %d, reduce_out = %d, sum_recvd = %d\n", 
                      Sdf::PROC_ID, i, recvd, reduce_out, sum_recvd);
              fflush(Sdf::trace);
            }

            Sdf::barrier(cid1);
                //this is to protect chan destruction
            delete chan;
       }
};


MAIN(Root_proc, init_gid, Sdf::empty, Sdf::empty)

